/**
 * 
 */
package com.xpec.c4.service.group.transportlayer;

import java.lang.Thread.State;
import java.util.WeakHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.xpec.c4.game.common.Helper;
import com.xpec.c4.game.time.JobSchedule;
import com.xpec.c4.game.utility.GUID;

/**
 * 处理本地消息请求
 * 
 * @author lipeilin
 * @2011-3-18 下午02:18:59
 * 
 */

public class MessageTimer {

	public static final int POOL_SIZE = 10;

	/**
	 * 保留时间
	 */
	public static final long KEEP_ALIVE_TIME = 500;

	/**
	 * 队列的长度
	 */
	public static final int QUEUE_SIZE = 100000;

	/**
	 * 队列的流量控制
	 */
	public static final int QUEUE_FLOW_CONTROL = (int) (QUEUE_SIZE * 0.95);

	/**
	 * 处理的Timer
	 */
	private static final WeakHashMap<GUID, Worker> scheduleMap = new WeakHashMap<GUID, Worker>();

	/**
	 * 任务执行器,主要用于集群内的消息处理
	 */
	private static final JobSchedule jobExecutor = new JobSchedule(5);

	/**
	 * 任务执行线程,主要用于client的消息处理,因为同一client的消息处理要按序.
	 */
	private static final Worker[] schedules = new Worker[POOL_SIZE];
	
	private MessageTimer() {
	}

	public static void handleMessage(GroupMessage msg) {
		GUID id = msg.getSessionID();
		MessageTask task = new MessageTask(msg);

		if (id == null) {
			jobExecutor.execute(task);
		} else {
			Worker schedule = getSchedules(id);
			schedule.execute(task);
		}

	}

	private static Worker getSchedules(GUID id) {
		// 如果是null，则找一个任务最少的。
		if (id == null) {
			return getRandomTaskThread();
		}
		Worker schedule = scheduleMap.get(id);
		// 如果没有之前的处理线程，也找一个最小的
		if (schedule == null || schedule.getState() == State.TERMINATED) {
			schedule = getRandomTaskThread();
			scheduleMap.put(id, schedule);
		}
		return schedule;
	}

	private static Worker getMinTaskThread() {
		long min = Integer.MAX_VALUE;
		Worker executeJob = null;
		for (int i = 0; i < schedules.length; i++) {
			Worker job = schedules[i];
			// 如果处理器为null，或者已经结束，则创建新的
			if (job == null || job.getState() == State.TERMINATED) {
				job = new Worker();
				schedules[i] = job;
				return job;
			}
			long count = job.getTaskSize();
			// 如果是0个，直接就用它吧！
			if (count == 0) {
				return job;
			}
			// 找出最少的
			if (min > count) {
				min = count;
				executeJob = job;
			}
		}
		return executeJob;
	}

	private static Worker getRandomTaskThread() {
		int index = Helper.randomInt(0, schedules.length);
		Worker executeJob = schedules[index];
		// 如果处理器为null，或者已经结束，则创建新的
		if (executeJob == null || executeJob.getState() == State.TERMINATED) {
			executeJob = new Worker();
			schedules[index] = executeJob;
		}
		return executeJob;
	}

}

class Worker extends Thread {

	private static final Log log = LogFactory.getLog(Worker.class);

	/**
	 * 消息处理队列
	 */
	private volatile LinkedBlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>(
			MessageTimer.QUEUE_SIZE);

	/**
	 * 是否在停止线程
	 */
	private boolean stop = false;

	private static AtomicInteger count = new AtomicInteger(0);

	/**
	 * 
	 */
	public Worker() {
		super("MessageWorkThread-" + count.incrementAndGet());
	}

	@Override
	public void run() {

		Runnable task;

		while (!stop) {
			while ((task = getTask()) != null) {
				task.run();
			}
		}
	}

	/**
	 * 执行任务
	 * 
	 * @param task
	 * 
	 * @2011-12-27 下午07:18:43
	 */
	public void execute(Runnable task) {
		int taskSize = getTaskSize();
		if (taskSize > MessageTimer.QUEUE_FLOW_CONTROL) {
			log.error(this + " message flow control!:message queue size:"
					+ taskSize + " task:" + task);
			return;
		}
		queue.add(task);

		State state = getState();

		switch (state) {
		case NEW:
			try {
				this.start();
			} catch (Exception e) {
				log.error(this + " start error!", e);
			}
			break;
		default:
			break;
		}

	}

	/**
	 * 获得对应的任务
	 * @return
	 */
	private Runnable getTask() {
		Runnable runnable = null;

		runnable = (Runnable) queue.poll();
		
		//没有的话，就等待输出
		if (runnable == null) {
			try {
				runnable = (Runnable) queue.poll(MessageTimer.KEEP_ALIVE_TIME,
						TimeUnit.MILLISECONDS);
			} catch (InterruptedException e) {
			}
		}
		return runnable;
	}

	/**
	 * 返回任务的数量
	 * 
	 * @return
	 * 
	 * @2011-12-27 下午07:18:34
	 */
	public int getTaskSize() {
		return queue.size();
	}
}
